package com.heima.kafka.simple;

import org.apache.kafka.clients.producer.*;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 生产者快速入门
 */
public class ProducerStart {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //2.设置连接信息
        Properties properties=new Properties();
            //指定kafka连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.104:9092");
        //设置发送消息的key和value均进行字符串序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        properties.put(ProducerConfig.RETRIES_CONFIG,5);
        //1.创建生产者对象
        KafkaProducer<String,String> kafkaProducer=new KafkaProducer<String, String>(properties);

        /**
         *  3.发送消息
         *  ProducerRecord(String topic, K key, V value)
         *  参数1表示发送消息的主题
         *  参数2表示消息的key  一定要是字符串类型
         *  参数3表示消息的内容   一定要是字符串类型
         */
        for (int i = 0; i <5 ; i++) {
            if(i%2==0){
                ProducerRecord<String, String> record=new ProducerRecord<>("stream-send-topic","key001","hello kafka");
                kafkaProducer.send(record);
            }else{
                ProducerRecord<String, String> record=new ProducerRecord<>("stream-send-topic","key001","hello itcast");
                kafkaProducer.send(record);
            }
        }

         /*  Future<RecordMetadata> future = kafkaProducer.send(record);
        RecordMetadata recordMetadata = future.get();
        System.out.println(recordMetadata.offset());*/
        //异步发送
       /* kafkaProducer.send(record, new Callback() {
            *//**
             * 发送完成之后
             * @param metadata
             * @param exception
             *//*
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                //如果发生了异常的话，
                if(exception!=null){
                    //则可以执行重试机制
                    System.out.println("发生异常了"+exception.getMessage());
                }
            }
        });*/


        /**
         * 关闭消息流
         * 注意，一定要关闭，否则消息没有发送成功
         */
        kafkaProducer.close();
    }
}
